package com.chukun.flink.stream.action.fraud.rules.functions;

import com.chukun.flink.stream.action.fraud.rules.utils.JsonMapper;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author chukun
 * @version 1.0.0
 * @description json解析器
 * @createTime 2022年05月22日 15:30:00
 */
public class JsonDeserializer<T> extends RichFlatMapFunction<String, T> {

    private static final Logger logger = LoggerFactory.getLogger(JsonDeserializer.class);

    private JsonMapper<T> jsonMapper;

    private final Class<T> targetClass;

    public JsonDeserializer(Class<T> targetClass) {
        this.targetClass = targetClass;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        jsonMapper = new JsonMapper<>(targetClass);
    }

    @Override
    public void flatMap(String value, Collector<T> collector) throws Exception {
        try {
            T parsed = jsonMapper.fromString(value);
            collector.collect(parsed);
        } catch (Exception e) {
            logger.warn("Failed parsing {}, dropping it: ", targetClass, e);
        }
    }


}
